//
// Created by juno on 2021/9/7.
//

#include <pthread.h>
#include <libwebsockets.h>
#include <json.h>

#include "ws_server.h"
#include "message.h"
#include "common.h"

static void *thread_spam(void *d)
{
    struct per_host_data *host = d;
    size_t n = 0;
    int i = 0;
    uint32_t tail;
    struct session_context ctx = {0};
    json_object *obj = NULL;
    int ret;
    pthread_t pid = pthread_self();

    do {
        if (!host->list)
            goto wait;

        lws_start_foreach_llp(struct per_session_data **, ppsession, host->list)
        {
            struct per_session_data *session = *ppsession;
            pthread_mutex_lock(&session->rx_lock);
            n = lws_ring_get_count_waiting_elements(session->rx_ring, &session->rx_tail);

            if (n > 0)
            {
                for (i = 0; i < n; i++)
                {
                    const struct msg *msg = lws_ring_get_element(session->rx_ring, &session->rx_tail);
                    if (!msg)
                        continue;

                    memset(&ctx, 0, sizeof(struct session_context));
                    ctx.wsi = session->wsi;
                    ctx.host = host;
                    ctx.session = session;

                    json_tokener *tok = json_tokener_new();
                    if (!tok)
                        goto next;

                    if (msg->len){
                        lwsl_debug("msg:\n%s\nlen:%lu\n", msg->payload, msg->len);
                        //insert_message(&ctx, 0,(const char *)msg->payload);
                        obj = json_tokener_parse_ex(tok, (char *)msg->payload, (int)msg->len);
                    }

                    json_tokener_free(tok);
                    next:
                    lws_ring_consume(session->rx_ring, &session->rx_tail, NULL, 1);
                    uint32_t oldest = lws_ring_get_oldest_tail(session->rx_ring);

                    if (oldest != session->rx_tail)
                    {
                        uint32_t _oldest = 0;
                        size_t _n = 0;
                        _oldest = session->rx_tail;
                        size_t _m = lws_ring_get_count_waiting_elements(session->rx_ring, &session->rx_tail);

                        if (_m >= _n)
                        {
                            _n = _m;
                            _oldest = session->rx_tail;
                        }
                        lws_ring_update_oldest_tail(session->rx_ring, _oldest);
                    }

                    if (obj)
                    {
                        process_message(&ctx, obj);
                        json_object_put(obj);           //TODO message process
                    }
                }
            }
            pthread_mutex_unlock(&session->rx_lock);
        }lws_end_foreach_llp(ppsession, list);
    wait:
        usleep(10000);
    }while(!host->finished);

    pthread_exit(NULL);
    lwsl_user("spam thread exit\n");
    return NULL;
}


int ws_server_msg_cb(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len)
{
    struct per_session_data *session = user;
    struct per_host_data *host = lws_protocol_vh_priv_get(lws_get_vhost(wsi), lws_get_protocol(wsi));
    const struct lws_protocol_vhost_options *pvo;
    int ret = 0;
    char buf[30] = {0};

    switch (reason) {
        case LWS_CALLBACK_PROTOCOL_INIT:
            host = lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi), lws_get_protocol(wsi), sizeof(struct per_host_data));
            pthread_mutex_init(&host->lock, NULL);

            pvo = lws_pvo_search(
                    (const struct lws_protocol_vhost_options *)in,
                            "chart"
                    );

            if (!pvo || !pvo->value)
            {
                lwsl_err("%s: Can't find \"chart\" pvo\n", __func__);
                return 1;
            }

            host->config = pvo->value;
            host->proto = lws_get_protocol(wsi);
            host->ctx = lws_get_context(wsi);
            host->host = lws_get_vhost(wsi);

            for (int n = 0; n < LWS_ARRAY_SIZE(host->pthread_spam); n++)
            {
                if (pthread_create(&host->pthread_spam[n], NULL, thread_spam, host))
                {
                    lwsl_err("Thread create failed\n");
                    ret = 1;
                    goto init_fail;
                }
            }

            break;
        case LWS_CALLBACK_PROTOCOL_DESTROY:
        init_fail:
            host->finished = 1;
            for (int n = 0; n < LWS_ARRAY_SIZE(host->pthread_spam); n++)
            {
                if (host->pthread_spam[n])
                {
                    lwsl_user("waiting for spam thread %d %#lx exit\n", n, host->pthread_spam[n]);
                    pthread_join(host->pthread_spam[n], NULL);
                }
                pthread_mutex_destroy(&host->lock);
            }
            break;
        case LWS_CALLBACK_ESTABLISHED:
        {
            lwsl_user("connecting\n");
            if (lws_hdr_copy(wsi, buf, sizeof(buf) -1,WSI_TOKEN_GET_URI) > 0)
                if (strncmp(buf, "/chart", 6))
                    break;

            session->rx_ring = lws_ring_create(sizeof(struct msg), 8, destroy_message);
            session->tx_ring = lws_ring_create(sizeof(struct msg), 8, destroy_message);

            pthread_mutex_init(&session->rx_lock, NULL);
            pthread_mutex_init(&session->tx_lock, NULL);

            session->publishing = true;
            session->wsi = wsi;
            gen_token(&session->token);
            session->rx_tail = lws_ring_get_oldest_tail(session->rx_ring);
            session->tx_tail = lws_ring_get_oldest_tail(session->tx_ring);

            pthread_mutex_lock(&host->lock);
            lws_ll_fwd_insert(session, list, host->list);
            pthread_mutex_unlock(&host->lock);
            lwsl_user("connected\n");
        }
            break;
        case LWS_CALLBACK_CLOSED:
            free(session->token);
            lws_ll_fwd_remove(struct per_session_data, list, session, host->list);
            break;

        case LWS_CALLBACK_SERVER_WRITEABLE:
        {
            lwsl_debug("LWS_CALLBACK_SERVER_WRITEABLE\n");

            if (!session->publishing)
                break;
            size_t n = lws_ring_get_count_waiting_elements(session->tx_ring, &session->tx_tail);
            if (!n)
                break;

            pthread_mutex_lock(&session->tx_lock);
            const struct msg *msg = lws_ring_get_element(session->tx_ring, &session->tx_tail);
            if (!msg)
                break;

            if (msg->len)
            {
                lwsl_debug("reply:%s\n", msg->payload);

                int m = lws_write(wsi, msg->payload + LWS_PRE, msg->len, LWS_WRITE_TEXT);
                if (m < msg->len)
                {
                    pthread_mutex_unlock(&session->tx_lock);
                    break;
                }
            }

            lws_ring_consume(session->tx_ring, &session->tx_tail, NULL, 1);
            uint32_t oldest = lws_ring_get_oldest_tail(session->tx_ring);
            if (oldest != session->tx_tail)
            {
                uint32_t _oldest = 0;
                size_t _n = 0;
                _oldest = session->tx_tail;
                size_t _m = lws_ring_get_count_waiting_elements(session->tx_ring, &session->tx_tail);

                if (_m >= _n)
                {
                    _n = _m;
                    _oldest = session->tx_tail;
                }
                lws_ring_update_oldest_tail(session->tx_ring, _oldest);
            }
            pthread_mutex_unlock(&session->tx_lock);
        }
            break;
        case LWS_CALLBACK_RECEIVE:
        {
            lwsl_user("recevie new msg\n");
            if (lws_frame_is_binary(wsi))
                break;

            if (!session->publishing)
                break;

            size_t n = lws_ring_get_count_free_elements(session->rx_ring);
            if (!n)
                break;
            pthread_mutex_lock(&session->rx_lock);
            struct msg msg = {0};
            msg.len = len;
            msg.payload = calloc(1, len +1);  //不能使用strdup, strdup会老拷贝in里边的所有内容,存在安全问题
            memcpy(msg.payload, in, len);

            if (1 != lws_ring_insert(session->rx_ring, &msg, 1))
            {
                destroy_message(&msg);
                goto unlock;
            }

            unlock:
            pthread_mutex_unlock(&session->rx_lock);
        }
            break;

        case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
            if (!host)
                break;

            lws_start_foreach_llp(struct per_session_data **,
                    ppss, host->list) {
                        if ((*ppss)->wsi)
                            lws_callback_on_writable((*ppss)->wsi);
            }lws_end_foreach_llp(ppss, list);
            break;

        case LWS_CALLBACK_FILTER_PROTOCOL_CONNECTION:
            if (lws_hdr_copy(wsi, buf, sizeof(buf) -1, WSI_TOKEN_GET_URI) > 0)
                ret = strncmp(buf, "/chart", 6);

            if (ret)
                return -1;
            break;
        default:
            break;
    }
    return 0;
}
